package com.yb.guigu.pick.activity.runner;

import com.yb.guigu.pick.activity.annotation.RedisStreamListener;
import com.yb.guigu.pick.activity.interceptor.RedisStreamInterceptor;
import com.yb.guigu.pick.activity.listener.redis.DemoRedisStreamListener;
import com.yb.guigu.pick.activity.listener.redis.RedisStreamConsumer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.aop.framework.ProxyFactoryBean;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Primary;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.util.ErrorHandler;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;

/**
 * Copyright (C), 2022-2023, 姚兵
 * Author: 32210
 * Date: 2023/7/14 13:56
 * FileName: RedisStreamRunner
 * Description:
 */

@Slf4j
@Component
public class RedisStreamRunner implements CommandLineRunner, ApplicationContextAware {

    @Autowired
    RedisConnectionFactory redisConnectionFactory;

    @Resource
    ThreadPoolTaskExecutor threadPoolTaskExecutor;


    private ApplicationContext applicationContext;


    @Autowired
    StringRedisTemplate stringRedisTemplate;



    private StreamMessageListenerContainer container;

    @Override
    public void run(String... args) throws Exception {



        initContainer();

        createConsumers();

        startContainer();
    }


    private void initContainer(){
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> containerOptions = StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                .builder()
                // 一次性最多拉取多少条消息
                .batchSize(10)
                // 执行消息轮询的执行器
                .executor(this.threadPoolTaskExecutor)
                // 消息消费异常的handler
                .errorHandler(t -> {
                    // throw new RuntimeException(t);
//                        t.printStackTrace();
//                        log.error("[MQ handler exception] " + t.getMessage());
                })
                // 超时时间，设置为0，表示不超时（超时后会抛出异常）
                .pollTimeout(Duration.ZERO)
                // 序列化器
                .serializer(new StringRedisSerializer())
                .build();


        this.container = StreamMessageListenerContainer.create(redisConnectionFactory,
                containerOptions);
    }


    private void startContainer(){
        container.start();
    }

    private void createConsumers(){
        Map<String, Object> beansWithAnnotation =
                applicationContext.getBeansWithAnnotation(RedisStreamListener.class);
        for (Object listener : beansWithAnnotation.values()) {

            final RedisStreamListener annotation = AnnotationUtils.findAnnotation(listener.getClass(), RedisStreamListener.class);

            try {

                createConsumer((RedisStreamConsumer) listener,annotation);

            }catch (Exception e){
                log.error("e:{}",e);
                throw e;
            }
        }
    }
    private Subscription createConsumer(RedisStreamConsumer redisStreamConsumer, RedisStreamListener redisStreamListener){



        initConsumer(redisStreamListener.streamKey(), redisStreamListener.group());

        redisStreamConsumer.setGroup(redisStreamListener.group());
        redisStreamConsumer.setConsumer(redisStreamListener.consumer());
        redisStreamConsumer.setAutoAck(redisStreamListener.autoAck());

        return redisStreamListener.autoAck()
                ?
                container.receiveAutoAck(Consumer.from(redisStreamListener.group(), redisStreamListener.consumer()),StreamOffset.create(redisStreamListener.streamKey(), ReadOffset.lastConsumed()), redisStreamConsumer)
                :container.receive(Consumer.from(redisStreamListener.group(), redisStreamListener.consumer()),StreamOffset.create(redisStreamListener.streamKey(), ReadOffset.lastConsumed()), redisStreamConsumer);
    }


    private void initConsumer(String key,String group){


        try {
            StreamInfo.XInfoGroups groups = stringRedisTemplate.opsForStream().groups(key);
            Optional<StreamInfo.XInfoGroup> infoGroup = groups.stream().filter(item -> item.groupName().equals(group)).findFirst();
            if(!infoGroup.isPresent()){

                stringRedisTemplate.opsForStream().createGroup(key,ReadOffset.from("0"),group);
            }
        }catch (Exception e){
            stringRedisTemplate.opsForStream().createGroup(key,ReadOffset.from("0"),group);
        }

    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext=applicationContext;
    }


    @PostConstruct
    public void destroy(){
       container.stop();
    }
}
